package com.vite.sse.service;

import com.vite.common.constant.CommonConstants;
import com.vite.common.core.domain.entity.vo.NoticeUnReadVo;
import com.vite.system.service.ISysNoticeService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

@Slf4j
@Service
@RequiredArgsConstructor
public class SseService {

    // 存储用户与SSE连接的映射
    private static final Map<String, SseEmitter> userEmitterMap = new ConcurrentHashMap<>();

    // 存储用户心跳任务的映射
    private static final Map<String, ScheduledFuture<?>> userHeartbeatMap = new ConcurrentHashMap<>();

    // SSE连接超时时间
    private static final long TIMEOUT = TimeUnit.MINUTES.toMillis(30);

    private static final String eventName = "notice_unRead";

    private static final int PLUS_NUM = 1;
    private static final int REDUCE_NUM = -1;

    // 公告服务
    private final ISysNoticeService noticeService;

    // 定时线程池
    private final ScheduledExecutorService scheduledExecutorService;

    // 业务线程池
    private final ThreadPoolTaskExecutor threadPoolTaskExecutor;

    /**
     * 建立SSE连接
     */
    public SseEmitter connect(String userName) {
        if (userEmitterMap.containsKey(userName)) {
            return userEmitterMap.get(userName);
        }
        // 创建新的SSE连接
        SseEmitter emitter = new SseEmitter(TIMEOUT);
        // 注册连接关闭回调
        emitter.onCompletion(() -> {
            userEmitterMap.remove(userName);
            cancelHeartbeat(userName);
            log.info("SSE连接关闭: {}", userName);
        });
        emitter.onTimeout(() -> {
            log.warn("SSE连接超时: {}", userName);
            emitter.complete();
            userEmitterMap.remove(userName);
            cancelHeartbeat(userName);
        });
        // 发送初始连接确认消息
        startConnect(userName,emitter);
        // 启动心跳机制
        startHeartbeat(userName, emitter);
        userEmitterMap.put(userName, emitter);
        log.info("SSE连接建立: {}", userName);
        return emitter;
    }

    /**
     * 发送连接成功消息
     */
    private void startConnect(String userName, SseEmitter emitter) {
        try {
            emitter.send(SseEmitter.event().name("connect").data("连接成功"));
        } catch (IOException e) {
            log.error("发送初始连接消息失败: {}", userName);
            emitter.complete();
        }
    }

    /**
     * 启动心跳机制
     */
    private void startHeartbeat(String userName, SseEmitter emitter) {
        ScheduledFuture<?> heartbeatFuture = scheduledExecutorService.scheduleAtFixedRate(() -> {
            try {
                // 发送心跳
                emitter.send(SseEmitter.event().name("heartbeat").data("{}"));
                log.debug("发送SSE心跳给用户: {}", userName);
            } catch (Exception e) {
                log.error("用户{}心跳发送失败，将取消心跳任务", userName);
                // 异常时主动取消任务并清理连接
                cancelHeartbeat(userName);
                emitter.complete();
                userEmitterMap.remove(userName);
            }
        }, 30, 30, TimeUnit.SECONDS);
        userHeartbeatMap.put(userName, heartbeatFuture);
    }

    /**
     * 取消用户的心跳任务
     */
    private void cancelHeartbeat(String userName) {
        ScheduledFuture<?> future = userHeartbeatMap.remove(userName);
        if (future != null && !future.isCancelled()) {
            future.cancel(false);
            log.debug("用户{}的心跳任务已取消", userName);
        }
    }

    /**
     * 消息单推
     */
    public void sendToOne(String userName, NoticeUnReadVo data) {
        SseEmitter emitter = userEmitterMap.get(userName);
        if (emitter == null) {
            log.warn("用户未建立SSE连接: {}", userName);
            return;
        }
        try {
            emitter.send(SseEmitter.event()
                    .name(eventName)
                    .data(data));
            log.info("发送SSE消息给用户: {}", userName);
        } catch (IOException e) {
            log.error("发送SSE消息失败: {}", userName);
            emitter.complete();
            userEmitterMap.remove(userName);
            cancelHeartbeat(userName);
        }
    }

    /**
     * 已读单发
     */
    public void redoNotice(String noticeType,String userName) {
        sendToOne(userName, getNoticeUnReadVo(noticeType,REDUCE_NUM));
    }

    /**
     * 新公告群发
     */
    public void sendNewNotice(String noticeType,String operatorName) {
        // 群发异步执行
        threadPoolTaskExecutor.execute(() ->userEmitterMap.keySet().stream()
                .filter(userName -> !operatorName.equals(userName)) // 过滤掉操作人
                .forEach(userName -> sendToOne(userName, getNoticeUnReadVo(noticeType,PLUS_NUM)))
        );
    }

    /**
     * 重新推送公告
     */
    public void sendRePush(Long noticeId, String noticeType,String operatorName) {
        List<String> onlineUsers = List.copyOf(userEmitterMap.keySet());
        if (onlineUsers.isEmpty()) {
            return;
        }
        List<String> readUsers = noticeService.getReadUserNameList(noticeId, onlineUsers);
        readUsers.remove(operatorName); // 排除操作人
        // 异步执行
        threadPoolTaskExecutor.execute(() -> onlineUsers.stream()
                .filter(readUsers::contains)
                .forEach(userName -> sendToOne(userName, getNoticeUnReadVo(noticeType,PLUS_NUM)))
        );
    }

    /**
     * 构建通知数据对象
     */
    private NoticeUnReadVo getNoticeUnReadVo(String noticeType,int num) {
        NoticeUnReadVo unReadVo = new NoticeUnReadVo().setNoticeType(noticeType);
        if (CommonConstants.NOTICE_TYPE.NOTICE_TZ.equals(noticeType)) {
            unReadVo.setTzUnReadNum(num);
        } else {
            unReadVo.setGgUnReadNum(num);
        }
        return unReadVo;
    }

}
